/*
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you under the Apache License, Version 2.0 (the
 *  "License"); you may not use this file except in compliance
 *  with the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 *
 */
package com.google.code.bigyang.nio;

import java.io.IOException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import com.google.code.bigyang.base.SessionState;
import com.google.code.bigyang.config.ExtendConfiguration;
import com.google.code.bigyang.log.Log;
import com.google.code.bigyang.util.ThreadUtil;

public class SelectorSet {

	protected SelectorManager[] select = new SelectorManager[ExtendConfiguration.getSelectorNum()];
	
	protected ConcurrentLinkedQueue<NioSessionEvent> queue = new ConcurrentLinkedQueue<NioSessionEvent>();
	
	private AtomicInteger index = new AtomicInteger(0);
	
	public void start() {
		for(int i=0;i<select.length;i++) {
			select[i] = new SelectorManager();
		}
		
		for(int i=0;i<select.length;i++) {
			ThreadUtil.createThread(select[i], "selector-"+(i+1));
		}
	}
	
	public void registerEvent(NioSessionEvent event) {
		SelectorManager manager = select[ Math.abs(index.incrementAndGet()) % select.length];
		manager.addQueue(event);
	} 
	
	protected class SelectorManager implements Runnable {
		private Selector selector;
		protected ConcurrentLinkedQueue<NioSessionEvent> queue = new ConcurrentLinkedQueue<NioSessionEvent>();
		
		public SelectorManager() {
			try {
				selector = Selector.open();
			} catch (IOException e) {
				Log.error(e.getMessage(),e);
			}
		}
		
		public void addQueue(NioSessionEvent event) {
			queue.add(event);
		}
		
		protected void handleEvent() throws InterruptedException, IOException {
			NioSessionEvent event = queue.poll();
			if(event != null) {
				SessionState state = event.getState();
				NioSession session = event.getSession();
				session.handleEvent(state, selector);
			}
		}
		
		protected void handleKeys() throws IOException {
			selector.select( 3 * 1000 );
			Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
			while(iter.hasNext()) {
				SelectionKey key = iter.next();
				iter.remove();
				
				if(!key.isValid()) {
					
				}
				
				if(key.isValid() && key.isReadable()) {
					NioSession session = (NioSession)key.attachment();
					key.interestOps(key.interestOps() & ~SelectionKey.OP_READ);
					new Thread(new NioTask(session),"task").start();
				} 
				
				else if(key.isValid() && key.isWritable()) {
					
				}
				
				else if(key.isValid() && key.isConnectable()) {
					
				}
			}
		}
		
		protected void handleTimeout() {
			
		}
		
		public void run() {
			while(true) {
				try {
					
					//1. 从队列中取session
					// session.handleEvent()
					handleEvent();
					
					
					//2.遍历  key，处理
					// session.handleEvent()
					handleKeys();
					
					
					//3.超时处理
					handleTimeout();
					
					
					
				} catch (InterruptedException e) {
					Log.error(e.getMessage(),e);
				} catch (ClosedChannelException e) {
					Log.error(e.getMessage(),e);
				} catch (IOException e) {
					Log.error(e.getMessage(),e);
				}
			}
		}
		
	}
	
}
